#ifndef __M_BROKER_H__
#define __M_BROKER_H__
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "host.hpp"
#include "connection.hpp"
#include "consumer.hpp"
#include "../mqcommon/threadpool.hpp"
#include <iostream>
#include <functional>

namespace mmq
{
    #define DBFILE "/meta.db"
    #define HOSTNAME "MyVirtualHost"
    class Server
    {
    private:
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    public:
        Server(int port, const std::string &basedir) :
            _server(&_loop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::Option::kReusePort),
            _dispatcher(std::bind(&Server::OnUnkonwnMessage, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
            _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
            _virutal_host(std::make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE)),
            _consumer_manager(std::make_shared<ConsumerManager>()),
            _connection_manager(std::make_shared<ConnectManager>()),
            _threadpool(std::make_shared<Threadpool>())
        {
            // 初始化队列的消费者管理句柄
            QueueMap qm = _virutal_host->AllQueues();
            for(auto &queue : qm)
            {
                _consumer_manager->InitQueueConsumer(queue.first);
            }
            // 注册 业务请求处理函数
            _dispatcher.registerMessageCallback<mmq::OpenChannelRequest>(std::bind(&Server::OnOpenChannel, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::CloseChannelRequest>(std::bind(&Server::OnCloseChannel, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::DeclareExchangeRequest>(std::bind(&Server::OnDeclareExchange, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::DeleteExchangeRequest>(std::bind(&Server::OnDeleteExchange, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::DeclareQueueRequest>(std::bind(&Server::OnDeclareQueue, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::DeleteQueueRequest>(std::bind(&Server::OnDeleteQueue, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::QueueBindRequest>(std::bind(&Server::OnQueueBind, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::QueueUnBindRequest>(std::bind(&Server::OnQueueUnBind, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::BasicPublishRequest>(std::bind(&Server::OnBasicPublish, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::BasicAckRequest>(std::bind(&Server::OnBasicAck, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::BasicConsumeRequest>(std::bind(&Server::OnBasicConsume, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mmq::BasicCancelRequest>(std::bind(&Server::OnBasicCancel, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            
            _server.setConnectionCallback(std::bind(&Server::OnConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        }

        void Start()    
        {
            _server.start();
            _loop.loop();
        }

    private:
        // 打开信道
        void OnOpenChannel(const muduo::net::TcpConnectionPtr &conn, const OpenChannelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("打开信道失败，找不到连接");
                conn->shutdown();
                return;
            }
            self_conn->OpenChannel(message);
        }
        // 关闭信道
        void OnCloseChannel(const muduo::net::TcpConnectionPtr &conn, const CloseChannelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("关闭信道失败，找不到连接");
                conn->shutdown();
                return;
            }
            self_conn->CloseChannel(message);
        }
        // 声明交换机
        void OnDeclareExchange(const muduo::net::TcpConnectionPtr &conn, const DeclareExchangeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("声明交换机失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("声明交换机失败，找不到信道");
                return;
            }
            channel->DeclareExchange(message);
        }
        // 删除交换机
        void OnDeleteExchange(const muduo::net::TcpConnectionPtr &conn, const DeleteExchangeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("删除交换机失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("删除交换机失败，找不到信道");
                return;
            }
            channel->DeleteExchange(message);
        }
        // 声明队列
        void OnDeclareQueue(const muduo::net::TcpConnectionPtr &conn, const DeclareQueueRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("声明队列失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("声明队列失败，找不到信道");
                return;
            }
            channel->DeclareQueue(message);
        }
        // 删除队列
        void OnDeleteQueue(const muduo::net::TcpConnectionPtr &conn, const DeleteQueueRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("删除队列失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("删除队列失败，找不到信道");
                return;
            }
            channel->DeleteQueue(message);
        }
        // 队列绑定
        void OnQueueBind(const muduo::net::TcpConnectionPtr &conn, const QueueBindRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("队列绑定失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("队列绑定失败，找不到信道");
                return;
            }
            channel->QueueBind(message);
        }
        // 队列解绑
        void OnQueueUnBind(const muduo::net::TcpConnectionPtr &conn, const QueueUnBindRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("队列解绑失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("队列解绑失败，找不到信道");
                return;
            }
            channel->QueueUnBind(message);
        }
        // 消息发布
        void OnBasicPublish(const muduo::net::TcpConnectionPtr &conn, const BasicPublishRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("消息发布失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("消息发布失败，找不到信道");
                return;
            }
            channel->BasicPublish(message);
        }
        // 消息确认
        void OnBasicAck(const muduo::net::TcpConnectionPtr &conn, const BasicAckRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("消息确认失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("消息确认失败，找不到信道");
                return;
            }
            channel->BasicAck(message);
        }
        // 队列消息订阅
        void OnBasicConsume(const muduo::net::TcpConnectionPtr &conn, const BasicConsumeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("队列消息订阅失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("队列消息订阅失败，找不到信道");
                return;
            }
            channel->BasicConsume(message);
        }
        // 队列消息取消订阅
        void OnBasicCancel(const muduo::net::TcpConnectionPtr &conn, const BasicCancelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr self_conn = _connection_manager->GetConnection(conn);
            if(self_conn.get() == nullptr)
            {
                DLOG("队列消息取消订阅失败，找不到连接");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = self_conn->GetChannel(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("队列消息取消订阅失败，找不到信道");
                return;
            }
            channel->BasicCancel(message);
        }

        void OnConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if(conn->connected())
            {
                _connection_manager->NewConnection(_consumer_manager, _virutal_host, _codec, conn, _threadpool);
            }
            else
            {
                _connection_manager->DelConnection(conn);
            }
        }

        void OnUnkonwnMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            std::cout << "onUnkonwnMessage: " << message->GetTypeName() << std::endl;
            conn->shutdown(); // 直接断开连接
        }

    private:
        muduo::net::EventLoop _loop;
        muduo::net::TcpServer _server; // 服务器对象
        ProtobufDispatcher _dispatcher; // 请求分发器对象--要向其中注册请求处理函数
        ProtobufCodecPtr _codec; // protobuf协议处理器--针对收到的请求数据进行prorobuf协议处理
        VirtualHost::ptr _virutal_host;
        ConsumerManager::ptr _consumer_manager;
        ConnectManager::ptr _connection_manager;
        Threadpool::ptr _threadpool;
    };

}


#endif